/****************************************************************************/
/* Dag description                                                          */
/*                                                                          */
/* Author(s):                                                               */
/* - Abdelkader AMAR (Abdelkader.Amar@ens-lyon.fr)                          */
/* - Benjamin ISNARD (Benjamin.Isnard@ens-lyon.fr)                          */
/*                                                                          */
/* $LICENSE$                                                                */
/****************************************************************************/
/* $Id: Dag.cc,v 1.23 2008/12/09 12:12:14 bisnard Exp $
 * $Log: Dag.cc,v $
 * Revision 1.23  2008/12/09 12:12:14  bisnard
 * added reference to FWorkflow
 *
 * Revision 1.22  2008/12/02 10:11:41  bisnard
 * modified node precedence check
 *
 * Revision 1.21  2008/10/29 11:00:14  bisnard
 * bug correction in dag destructor
 *
 * Revision 1.20  2008/10/22 09:29:00  bisnard
 * replaced uint by standard type
 *
 * Revision 1.19  2008/10/20 08:01:12  bisnard
 * removed createNode method from NodeSet class
 *
 * Revision 1.18  2008/10/14 13:31:01  bisnard
 * new class structure for dags (DagNode,DagNodePort)
 *
 * Revision 1.17  2008/09/30 15:32:53  bisnard
 * - using simple port id instead of composite ones
 * - dag nodes linking refactoring
 * - prevNodes and nextNodes data structures modified
 * - prevNodes initialization by Node::setNodePredecessors
 *
 * Revision 1.16  2008/09/30 09:23:29  bisnard
 * removed diet profile initialization from DagWfParser and replaced by node methods initProfileSubmit and initProfileExec
 *
 * Revision 1.15  2008/09/19 13:11:07  bisnard
 * - added support for containers split/merge in workflows
 * - added support for multiple port references
 * - profile for node execution initialized by port (instead of node)
 * - ports linking managed by ports (instead of dag)
 *
 * Revision 1.14  2008/07/17 12:19:18  bisnard
 * Added dag cancellation method
 *
 * Revision 1.13  2008/07/11 07:56:01  bisnard
 * provide list of failed nodes in case of cancelled dag
 *
 * Revision 1.12  2008/07/08 11:15:58  bisnard
 * Correct dag/node destruction with nodequeues
 *
 * Revision 1.11  2008/06/25 10:07:12  bisnard
 * - removed debug messages
 * - Node index in wf_response stored in Node class (new attribute submitIndex)
 *
 * Revision 1.10  2008/06/19 10:17:13  bisnard
 * new attribute to store DAG start time
 *
 * Revision 1.9  2008/06/18 14:58:57  bisnard
 * bug in updateDelayRec
 *
 * Revision 1.8  2008/06/02 08:35:39  bisnard
 * Avoid MaDag crash in case of client-SeD comm failure
 *
 * Revision 1.7  2008/06/01 15:50:59  rbolze
 * less verbose
 *
 * Revision 1.6  2008/06/01 09:18:43  rbolze
 * remove myreqID attribute from the RunnableNode
 * add getReqID() method which return the reqID stored in the diet_profile_t
 * add 2 methods: showDietReqID and getAllDietReqID
 * this is use to send feedback through logservice
 *
 * Revision 1.5  2008/05/16 12:32:10  bisnard
 * API function to retrieve all workflow results
 *
 * Revision 1.4  2008/04/28 12:14:59  bisnard
 * new NodeQueue implementation for FOFT
 * nodes sorting done by Dag instead of scheduler
 * method to propagate delay at execution (FOFT)
 *
 * Revision 1.3  2008/04/21 14:36:59  bisnard
 * use nodeQueue to manage multiwf scheduling
 * use wf request identifer instead of dagid to reference client
 * renamed WfParser as DagWfParser
 *
 * Revision 1.2  2008/04/14 09:10:40  bisnard
 *  - Workflow rescheduling (CltReoMan) no longer used with MaDag v2
 *  - AbstractWfSched and derived classes no longer used with MaDag v2
 *
 * Revision 1.1  2008/04/10 08:38:50  bisnard
 * New version of the MaDag where workflow node execution is triggered by the MaDag agent and done by a new CORBA object CltWfMgr located in the client
 *
 * Revision 1.6  2006/11/08 15:18:58  aamar
 * Complete get_all_results method to handle matrix and file types.
 *
 * Revision 1.5  2006/11/06 11:56:42  aamar
 * Workflow support:
 *    - Adding get_file_output and get_matrix_output (to get workflow file
 *      and matrix results)
 *
 * Revision 1.4  2006/10/26 13:57:53  aamar
 * Replace cout/cerr by debug macro TRACE_XXX
 *
 * Revision 1.3  2006/10/20 08:40:02  aamar
 * Adding the following functions to the class:
 *    - isEnabled(); setEnable; and getRef
 * Implement the new (modified) function getRemainingDag
 *
 * Adding the following function to DAG class
 *   - getNodes; getId; setId; getInputNodes; getOutputNodes();
 *   getAllProfiles(); setAsTemp; getEstMakespan();
 *
 * Revision 1.2  2006/07/10 11:08:12  aamar
 * - Adding the toXML function that return the DAG XML
 * representation
 * - Adding reordering (rescheduling) management
 *
 * Revision 1.1  2006/04/14 13:46:51  aamar
 * Direct acyclic graph class (source).
 *
 ****************************************************************************/

#include "debug.hh"

#include "Dag.hh"
#include "DagWfParser.hh"

using namespace std;

NodeSet::NodeSet() { }
NodeSet::~NodeSet() { }

Dag::Dag() : myId(""), tmpDag(false), estDelay(0), cancelled(false),
             myWf(NULL) {
}

/**
 * DAG destructor: will delete all the nodes of the dag (if not a temp dag)
 */
Dag::~Dag() {
  TRACE_TEXT (TRACE_ALL_STEPS,"~Dag() destructor ..." <<  endl);
  if (! this->tmpDag) {
    while (! nodes.empty() ) {
       DagNode * p = begin()->second ;
       nodes.erase( begin() ) ;
       delete p ;
    }
    nodes.clear();
  }
}

/**
 * set the dag id
 */
void
Dag::setId(const string& id) {
  this->myId = id;
}

/**
 * get the dag id
 */
const string&
Dag::getId() {
  return this->myId;
}

/**
 * Set the functional wf for which this dag is an instance
 */
void
Dag::setWorkflow(FWorkflow * wf) {
  this->myWf = wf;
}

/**
 * Get the functional wf
 */
FWorkflow *
Dag::getWorkflow() {
  return this->myWf;
}

/**
 * Get the node with given identifier (only node id, not the complete id)
 */
Node *
Dag::getNode(const string& nodeId) {
  return getDagNode(nodeId);
}

/**
 * Allocates a new node and add it to the dag
 */
DagNode*
Dag::createDagNode(const string& id) {
  if (this->getDagNode(id) != NULL)
    return NULL;
  DagNode* newDagNode = new DagNode(this, id);
  this->nodes[id] = newDagNode;
  return newDagNode;
}

/**
 * Get the dag node with given identifier (only node id, not the complete id)
 */
DagNode *
Dag::getDagNode(const string& nodeId) {
  map<string, DagNode*>::iterator p = this->nodes.find(nodeId);
  if ( p != this->nodes.end())
    return p->second;
  else
    return NULL;
}

/**
 * check the precedence between node *
 * this function checks the precedence between node; it doesn't *
 * link the WfPorts but it creates the list of predecessors of each node *
 */
void
Dag::checkPrec(NodeSet* contextNodeSet) throw (WfStructException) {
  TRACE_TEXT(TRACE_ALL_STEPS, "CHECKING DAG STRUCTURE START" << endl);
  for (map<string, DagNode * >::iterator p = nodes.begin( );
       p != nodes.end( );
       ++p ) {
    DagNode *node = (DagNode*) p->second;
    if (!node->setNodePrecedence(contextNodeSet))
      throw WfStructException(WfStructException::eUNKNOWN_NODE,
                              "Cannot set predecessor relationship for node "
                                  + node->getId());
  }
  // TODO use DFS to check there is no cycle
  TRACE_TEXT(TRACE_ALL_STEPS, "CHECKING DAG STRUCTURE END" << endl);
}

/**
 * link all ports of the dag *
 */
void
Dag::linkAllPorts() {
  TRACE_TEXT(TRACE_ALL_STEPS, "LINKING NODES START" << endl);
  for (map<string, DagNode*>::iterator p = nodes.begin();
       p != nodes.end();
       ++p) {
    DagNode * n = (DagNode*)(p->second);
    n->connectNodePorts();
  }
  TRACE_TEXT(TRACE_ALL_STEPS, "LINKING NODES END" << endl);
}

/**
 * return the size of the Dag (the nodes number)
 */
unsigned int
Dag::size() {
  return nodes.size();
}

/**
 * return an iterator on the first node *
 * (according to the map and not to the dag structure) *
 */
map <string, DagNode *>::iterator
Dag::begin() {
  return nodes.begin();
}

/**
 * return an iterator on the last node *
 * (according to the map and not to the dag structure) *
 */
map <string, DagNode *>::iterator
Dag::end() {
 return nodes.end();
}

/**
 * returns the XML description of the dag
 */
string
Dag::toXML() {
  string xml = "<dag>\n";
  for (map<string, DagNode*>::iterator p = begin();
       p != end();
       ++p) {
    xml += ((DagNode*) p->second)->toXML();
  }
  xml += "</dag>\n";
  return xml;
}

/**
 * printDietReqID the dag execution is completed
 *
 * this methods loops through all the nodes of the dag and getReqID of each node
 * @return reqID[]
 */
void
Dag::showDietReqID() {
  DagNode * dagNode = NULL;
  cout << "@@@@ BEGIN Dag::" <<__FUNCTION__  << "()" << endl;
  cout << "dag_id =" << this->myId << endl;
  for (map<string, DagNode *>::iterator p = nodes.begin();
       p != nodes.end();
       ++p) {
    dagNode = (DagNode*)p->second;
    //if ((dagNode) && !(dagNode->isDone()))
    cout << " dagNode->getPb() = "  << dagNode->getPbName() <<endl;
    cout << " dagNode->getProfile()->dietReqID =" << dagNode->getProfile()->dietReqID << endl;
  }
  cout << "@@@@ END Dag::" <<__FUNCTION__  << "()" << endl;
}
/**
 * printDietReqID the dag execution is completed
 *
 * this methods loops through all the nodes of the dag and getReqID of each node
 * @return diet_reqID_t[]
 */
vector<diet_reqID_t>
Dag::getAllDietReqID() {
	DagNode * dagNode = NULL;
	vector<diet_reqID_t> request_ids;
	//cout << "dag_id =" << this->myId << endl;
	for (map<string, DagNode *>::iterator p = nodes.begin();
		    p != nodes.end();
		    ++p) {
			    dagNode = (DagNode*)p->second;
			    request_ids.push_back(dagNode->getReqID());
			    //cout << "reqID ="<< dagNode->getReqID() << endl;
		    }
	return request_ids;
}

bool
Dag::isDone() {
  DagNode * dagNode = NULL;
  bool res = true;
  this->myLock.lock();  /** LOCK */
  for (map<string, DagNode *>::iterator p = nodes.begin();
       p != nodes.end();
       ++p) {
    dagNode = (DagNode*) p->second;
    if ((dagNode) && !(dagNode->isDone())) {
      res = false;
      break;
    }
  }
  this->myLock.unlock(); /** UNLOCK */
  return res;
}

/**
 * check if the dag execution is ongoing
 * (checks all nodes status and returns true if at least one node is running)
 */
bool
Dag::isRunning() {
  bool res = false;
  DagNode * dagNode = NULL;
  this->myLock.lock();  /** LOCK */
  for (map<string, DagNode *>::iterator p = nodes.begin();
       p != nodes.end();
       ++p) {
    dagNode = (DagNode*)p->second;
    if ((dagNode) && (dagNode->isRunning())) {
      res = true;
    }
  }
  this->myLock.unlock(); /** UNLOCK */
  return res;
}

/**
 * check if the dag execution is cancelled *
 */
bool
Dag::isCancelled() {
  return cancelled;
}

/**
 * local function to extract node and port name from composite
 * id (node#port) used by API
 */

DagNodeOutPort *
Dag::getOutputPort(const char* id) {
  string strId(id);
  string::size_type portSep = strId.find("#");
  string nodeId = strId.substr(0, portSep);
  string portId = strId.substr(portSep+1, strId.length()-portSep-1);
    TRACE_TEXT (TRACE_ALL_STEPS,
	      "\t" << "get_scalar_output : searching for " << id << endl);
  DagNode * n = getDagNode(nodeId);
  if (n != NULL) {
    DagNodeOutPort * outp = dynamic_cast<DagNodeOutPort*>(n->getPort(portId));
    if (outp != NULL) {
      return outp;
    } else {
      ERROR("Cannot get result for non-output port",0);
    }
  }
  ERROR("Output port not found", 0);
}

/**
 * Get a scalar result of the workflow *
 *
 * this methods loops through all dag nodes to find the output port identified
 * by id
 * @param id identifier of the node
 * @param value pointer to pointer to the value of the output port
 *
 * @todo unify the different methods used to retrieve an output for the different types
 */
int
Dag::get_scalar_output(const char * id,
                       void** value) {
  DagNodeOutPort * outp = getOutputPort(id);
  return diet_scalar_get(diet_parameter(outp->profile(),outp->getIndex()),
                         value, NULL);
}

/**
 * get a string result of the workflow *
 */
int
Dag::get_string_output(const char * id,
                       char** value) {
  DagNodeOutPort * outp = getOutputPort(id);
  return diet_string_get(diet_parameter(outp->profile(),outp->getIndex()),
                         value, NULL);
}

/**
 * Get a file result of the workflow
 */
int
Dag::get_file_output (const char * id,
                      size_t* size, char** path) {
  DagNodeOutPort * outp = getOutputPort(id);
  return diet_file_get(diet_parameter(outp->profile(), outp->getIndex()),NULL,
                       size, path);
}

/**
 * Get a matrix result of the workflow
 */
int
Dag::get_matrix_output (const char * id, void** value,
                        size_t* nb_rows, size_t *nb_cols,
                        diet_matrix_order_t* order) {
  DagNodeOutPort * outp = getOutputPort(id);
  return diet_matrix_get(diet_parameter(outp->profile(), outp->getIndex()),
                         value, NULL,
                         nb_rows, nb_cols, order);
}

/**
 * Get the nodes sorted according to their priority using insertion sort
 * (output vector must be deleted)
 */
std::vector<DagNode*>&
Dag::getNodesByPriority() {
  std::vector<DagNode*> * sorted_list = new std::vector<DagNode*>;
  DagNode * n1 = NULL;
  TRACE_TEXT (TRACE_ALL_STEPS, "Sorting dag nodes by priority" << endl);
  for (std::map <std::string, DagNode *>::iterator p = this->begin();
       p != this->end();
       p++) {
    n1 = (DagNode*)(p->second);
    // found where insert the node
    std::vector<DagNode*>::iterator p = sorted_list->begin();
    bool b = false;
    DagNode * n2 = NULL;
    while ((p != sorted_list->end()) && (!b)) {
      n2 = *p;
      if (n2->getPriority() < n1->getPriority())
        b = true;
      else
        p++;
    }
    sorted_list->insert(p, n1);
  }
  return *sorted_list;
}

/**
 * get all the results
 */
void
Dag::displayAllResults() {
  cout << "** RESULTS OF DAG " << getId() << " **" << endl;
  DagNode * n = NULL;
  for (map<string, DagNode *>::iterator p = nodes.begin();
       p != nodes.end();
       ++p) {
    n = (DagNode *)(p->second);
    if ((n != NULL) && (n->isAnExit())) {
      n->displayResults();
    }
  }
}

/**
 * free all persistent data used by the dag (includes intermediate and
 * final results)
 */
void
Dag::deleteAllResults() {
  DagNode * n = NULL;
  for (map<string, DagNode *>::iterator p = nodes.begin();
       p != nodes.end();
       ++p) {
    n = (DagNode *)(p->second);
    if (n != NULL) {
      n->freeProfileAndData();
    }
  }
}

/**
 * get the input nodes
 */
vector<DagNode *>
Dag::getInputNodes() {
  vector<DagNode *> v;
  DagNode * node = NULL;
  for (map<string, DagNode *>::iterator p = this->nodes.begin();
       p != this->nodes.end();
       p++) {
    node = (DagNode *)(p->second);
    if ((node != NULL) && (node->isAnInput()))
      v.push_back(node);
  }
  return v;
} // end getInputNodes

/**
 * set all input nodes as ready
 */
void
Dag::setInputNodesReady() {
  vector<DagNode *> inputs = this->getInputNodes();
  for (vector<DagNode *>::iterator p = inputs.begin();
       p != inputs.end();
       p++) {
    DagNode * node = (DagNode *) *p;
    node->setAsReady();
  }
}

/**
 * get the estimated makespan of the DAG
 * @deprecated
 */
// double
// Dag::getEstMakespan() {
//   double makespan = -1;
//   Node * n = NULL;
//   for (map<string, Node*>::iterator p = this->nodes.begin();
//        p != this->nodes.end();
//        ++p) {
//     n = (Node*)(p->second);
//     if ( (n != NULL) &&
// 	 (n->getEstCompTime() > makespan) )
//       makespan = n->getEstCompTime();
//   }
//   return makespan;
// }

/**
 * get the estimated earliest finish time of the DAG
 */
double
Dag::getEFT() {
  double EFT = -1;
  DagNode * n = NULL;
  for (map<string, DagNode*>::iterator p = this->nodes.begin();
       p != this->nodes.end();
       ++p) {
    n = (DagNode*)(p->second);
    if ( (n != NULL) &&
	 (n->getEstCompTime() > EFT) )
      EFT = n->getEstCompTime();
  }
  return EFT;
}

/**
 * set the start time of the DAG
 */
void
Dag::setStartTime(double time) {
  this->startTime = time;
}

/**
 * get the start time of the DAG
 */
double
Dag::getStartTime() {
  return this->startTime;
}

/**
 * get the estimated delay of the DAG
 */
double
Dag::getEstDelay() {
  return this->estDelay;
}

/**
 * set the estimated delay of the DAG
 * (updated by an exit node)
 * (can be increased or decreased)
 */
void
Dag::setEstDelay(double delay) {
  this->estDelay = delay;
  TRACE_TEXT (TRACE_ALL_STEPS, "Updated est. delay on DAG "
        << this->getId() << " : delay = " << delay << endl);
}

bool
Dag::updateDelayRec(DagNode * node, double newDelay) {
  bool res = true;
  this->myLock.lock();  /** LOCK */
  res = _updateDelayRec(node, newDelay);
  this->myLock.unlock();  /** UNLOCK */
  return res;
}

/**
 * RECURSIVE
 * updates the estDelay of node and propagates it to the successors
 * returns true if propagation worked well (no pb in getting delay info)
 */
bool
Dag::_updateDelayRec(DagNode * node, double newDelay) {
  bool res = true;
  if (newDelay > node->getEstDelay()) {
    // the node is/will be late compared to the last estimated delay
    // so the new delay must be propagated to the successors
    node->setEstDelay(newDelay);
    for (list<Node*>::iterator nextIter = node->nextNodesBegin();
         nextIter != node->nextNodesEnd();
         ++nextIter) {
      res = res && this->_updateDelayRec(dynamic_cast<DagNode*>(*nextIter), newDelay);
    }
  }
  else {
    TRACE_TEXT (TRACE_ALL_STEPS, "Delay estimate for node "
        << node->getCompleteId() << " unchanged (newDelay = "
        << newDelay << " / previous delay = " << node->getEstDelay()
        << ")" << endl);
  }
  return res;
}

/**
 * notify the dag of node execution failure (MADAG & CLIENT-SIDE)
 */
void
Dag::setNodeFailure(string nodeId) {
  this->cancelled = true;
  this->failedNodes.push_front(nodeId);
}

/**
 * get the list of failed nodes
 */
const std::list<string>&
Dag::getNodeFailureList() {
  return this->failedNodes;
}
/**
 * set the dag as cancelled
 * (used when failure happens after all nodes execution)
 */
void
Dag::setAsCancelled() {
  this->cancelled = true;
}


